package com.yanggu.flink.datastream_api.test

import com.yanggu.flink.datastream_api.accumulate_batch.ProcessTimeMiniBatchOperator
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness
import org.scalatest.BeforeAndAfter
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers

import java.util.concurrent.TimeUnit

class TestProcessTimeMiniBatchOperator extends AnyFlatSpec with Matchers with BeforeAndAfter {


  "测试定时器和snapshot同时调度" should "应该加锁, 只执行一个" in {
    val testHarness = new OneInputStreamOperatorTestHarness(new ProcessTimeMiniBatchOperator(10000L, 10))

    //初始化
    testHarness.open()

    testHarness.processElement(new StreamRecord[String]("test1"))

    //触发定时器操作

    testHarness.setProcessingTime(System.currentTimeMillis() + 20000L)

    testHarness.snapshot(1L, 1L)

    val list = testHarness.extractOutputValues()
    println(list)


  }

}
